package com.viknix.threadpool.manager.cluster.core;

import cn.hutool.core.collection.ConcurrentHashSet;
import com.viknix.threadpool.manager.cluster.constant.MemberMetaDataConstants;
import com.viknix.threadpool.manager.cluster.enums.NodeState;
import com.viknix.threadpool.manager.cluster.event.MembersChangeEvent;
import com.viknix.threadpool.manager.cluster.executor.GlobalExecutor;
import com.viknix.threadpool.manager.cluster.factory.LookupFactory;
import com.viknix.threadpool.manager.cluster.lookup.MemberLookup;
import com.viknix.threadpool.manager.cluster.util.AuthHeaderUtil;
import com.viknix.threadpool.manager.cluster.util.GenericType;
import com.viknix.threadpool.manager.cluster.util.MemberUtil;
import com.viknix.threadpool.manager.http.*;
import com.viknix.threadpool.manager.http.util.HttpUtils;
import com.viknix.threadpool.manager.server.dependence.constant.Commons;
import com.viknix.threadpool.manager.server.dependence.constant.Constants;
import com.viknix.threadpool.manager.server.dependence.exception.ServerProcessException;
import com.viknix.threadpool.manager.server.dependence.monitor.NotifyCenter;
import com.viknix.threadpool.manager.server.dependence.monitor.event.Event;
import com.viknix.threadpool.manager.server.dependence.monitor.listeners.Listener;
import com.viknix.threadpool.manager.server.dependence.result.RestResult;
import com.viknix.threadpool.manager.server.dependence.util.EnvUtil;
import com.viknix.threadpool.manager.server.dependence.util.ExceptionUtil;
import com.viknix.threadpool.manager.server.dependence.util.InetUtils;
import com.viknix.threadpool.manager.server.dependence.util.VersionUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.boot.web.context.WebServerInitializedEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.http.HttpStatus;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;
import javax.servlet.ServletContext;
import java.util.*;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * @Author: Dongqi
 * @Date: 2021/11/11 9:44
 * @Version 1.0
 * @Description: 集群节点管理中心，监听 WebServerInitializedEvent 事件
 * 1、管理所有集群
 * 2、管理健康实例
 * 3、缓存当前节点信息
 * 4、与集群中其他节点保持通信
 * 集群状态切换规律：当前节点启动，扫描 cluster.conf，将自身节点状态置为 up，其余所有节点状态置为 down
 * 启动后延时5s执行 infoReportTask 任务与其余节点建立通信，建立成功，其余节点状态置为up，当通信失败，对方节点状态置为 SUSPICIOUS，
 * 当失败次数超过阈值，对方节点状态置为 down，此后每延时2s会定期执行 infoReportTask 任务。
 */
@Slf4j
@Component(value = "serverMemberManager")
public class ServerMemberManager implements ApplicationListener<WebServerInitializedEvent> {

    private final TpmAsyncRestTemplate asyncRestTemplate = HttpClientBeanHolder
            .getTpmAsyncRestTemplate();

    /**
     * key：ip + : + port，value：Member
     * 集群节点列表。
     */
    private volatile ConcurrentSkipListMap<String, Member> serverList;

    /**
     * 此节点是否在集群列表中。
     */
    private volatile boolean isInIpList = true;

    /**
     * 端口号
     */
    private int port;

    /**
     * 本地节点的地址信息。
     */
    private String localAddress;

    /**
     * 寻址模式实例。
     */
    private MemberLookup lookup;

    /**
     * 自我成员对象。
     */
    private volatile Member self;

    /**
     * 存储节点状态为 UP 的节点
     */
    private volatile Set<String> memberAddressInfos = new ConcurrentHashSet<>();

    /**
     * 轮询向集群中除自身以外的所有其他节点广播当前节点的信息
     */
    private final MemberInfoReportTask infoReportTask = new MemberInfoReportTask();

    public ServerMemberManager(ServletContext servletContext) throws Exception {
        this.serverList = new ConcurrentSkipListMap<>();
        EnvUtil.setContextPath(servletContext.getContextPath());

        init();
    }

    protected void init() throws ServerProcessException {

        log.info(">>> 😊 Tpm-related cluster resource initialization");
        this.port = EnvUtil.getProperty("server.port", Integer.class, 3180);
        this.localAddress = InetUtils.getSelfIP() + ":" + port;
        // 根据本机ip地址和项目端口号构建当前节点的Member对象
        this.self = MemberUtil.singleParse(this.localAddress);
        // 设置扩展信息
        this.self.setExtendVal(MemberMetaDataConstants.VERSION, VersionUtils.version);
        // 将当前节点信息放入到 serverList 集合中
        serverList.put(self.getAddress(), self);

        // 将 MembersChangeEvent Publisher注册到 NotifyCenter
        registerClusterEvent();

        // 生产寻址策略，并初始化
        // 从磁盘读取配置信息，判断集群节点是否有变更
        // 如果有变更更新内存中节点信息，同时持久化新节点信息持久化到cluster.conf，发布 MembersChangeEvent 事件
        initAndStartLookup();

        if (serverList.isEmpty()) {
            throw new ServerProcessException(ServerProcessException.SERVER_ERROR, "cannot get serverlist, so exit.");
        }

        log.info(">>> 😊 The cluster resource is initialized");
    }

    /**
     * 生产并初始化寻址模式
     *
     * @throws ServerProcessException
     */
    private void initAndStartLookup() throws ServerProcessException {
        // 生产寻址策略
        this.lookup = LookupFactory.createLookUp(this);
        // 初始化寻址策略
        this.lookup.start();
    }

    /**
     * 切换到目标寻址模式
     *
     * @param name 目标寻址模式名称
     * @throws ServerProcessException
     */
    public void switchLookup(String name) throws ServerProcessException {
        this.lookup = LookupFactory.switchLookup(name, this);
        this.lookup.start();
    }

    private void registerClusterEvent() {

        // 注册节点更改事件发布器到 NotifyCenter
        NotifyCenter.registerToPublisher(MembersChangeEvent.class,
                EnvUtil.getProperty("tpm.member-change-event.queue.size", Integer.class, 128));

        // The address information of this node needs to be dynamically modified
        // when registering the IP change of this node
        NotifyCenter.registerSubscriber(new Listener<InetUtils.IPChangeEvent>() {
            @Override
            public void onEvent(InetUtils.IPChangeEvent event) {
                String newAddress = event.getNewIP() + ":" + port;
                ServerMemberManager.this.localAddress = newAddress;
                EnvUtil.setLocalAddress(localAddress);

                Member self = ServerMemberManager.this.self;
                self.setIp(event.getNewIP());

                String oldAddress = event.getOldIP() + ":" + port;
                ServerMemberManager.this.serverList.remove(oldAddress);
                ServerMemberManager.this.serverList.put(newAddress, self);

                ServerMemberManager.this.memberAddressInfos.remove(oldAddress);
                ServerMemberManager.this.memberAddressInfos.add(newAddress);
            }

            @Override
            public Class<? extends Event> subscribeType() {
                return InetUtils.IPChangeEvent.class;
            }
        });
    }

    /**
     * 节点信息更新，如果当前集群列表中不包含该节点信息，返回false
     *
     * @param newMember {@link Member} 信息变更的节点信息
     * @return 更新是否成功
     */
    public boolean update(Member newMember) {
        log.debug("member information update : {}", newMember);

        String address = newMember.getAddress();
        // 如果当前集群列表中不包含该节点信息，返回false
        if (!serverList.containsKey(address)) {
            return false;
        }

        // s是key，member指原来的集群节点信息
        serverList.computeIfPresent(address, (s, member) -> {
            // 如果节点状态已经是 DOWN，从 memberAddressInfos 集合中移除
            if (NodeState.DOWN.equals(newMember.getState())) {
                memberAddressInfos.remove(newMember.getAddress());
            }
            // 判断基本信息是否发生变化。如果基本信息有变化，则为true，否则为false
            boolean isPublishChangeEvent = MemberUtil.isBasicInfoChanged(newMember, member);
            // 更新最后修改时间
            newMember.setExtendVal(MemberMetaDataConstants.LAST_REFRESH_TIME, System.currentTimeMillis());
            // 将 newMember 数据拷贝到 oldMember 上
            MemberUtil.copy(newMember, member);
            if (isPublishChangeEvent) {
                // 成员基本数据发生变化，发布 MembersChangeEvent 事件
                notifyMemberChange();
            }
            // 返回更新完成后的节点信息
            return member;
        });
        return true;
    }

    public void notifyMemberChange() {
        NotifyCenter.publishEvent(MembersChangeEvent.builder().members(allMembers()).build());
    }

    /**
     * 该节点是否存在于集群中。
     *
     * @param address ip:port
     * @return 是否存在
     */
    public boolean hasMember(String address) {
        boolean result = serverList.containsKey(address);
        if (!result) {
            // 如果只传入IP信息，则需要进行模糊匹配
            for (Map.Entry<String, Member> entry : serverList.entrySet()) {
                if (StringUtils.contains(entry.getKey(), address)) {
                    result = true;
                    break;
                }
            }
        }
        return result;
    }

    public Member getSelf() {
        return this.self;
    }

    public Member find(String address) {
        return serverList.get(address);
    }

    /**
     * 返回这个集群的所有成员。
     *
     * @return 集群的所有成员
     */
    public Collection<Member> allMembers() {
        // 我们需要做一个拷贝，以免影响真实数据
        HashSet<Member> set = new HashSet<>(serverList.values());
        set.add(self);
        return set;
    }

    /**
     * 返回除自己以外所有集群成员。
     *
     * @return 除自己以外所有集群成员
     */
    public List<Member> allMembersWithoutSelf() {
        List<Member> members = new ArrayList<>(serverList.values());
        members.remove(self);
        return members;
    }

    /**
     * 判断集群节点是否有变更
     * 如果有变更更新内存中节点信息，同时持久化新节点信息持久化到cluster.conf，发布 MembersChangeEvent 事件
     *
     * @param members 所有集群节点的信息
     * @return
     */
    public boolean memberChange(Collection<Member> members) {

        if (members == null || members.isEmpty()) {
            return false;
        }

        // 判断参数中是否包含自身节点
        boolean isContainSelfIp = members.stream()
                .anyMatch(ipPortTmp -> Objects.equals(localAddress, ipPortTmp.getAddress()));

        if (isContainSelfIp) {
            isInIpList = true;
        } else {
            isInIpList = false;
            // 如果参数中不包含自身节点，将自身节点加入到参数中
            members.add(this.self);
            log.warn("[server list] self ip {} not in serverlist {}", self, members);
        }

        // TODO 第一次启动的时候一定会重写磁盘，serverList.size() == 1
        // 如果新旧集群的数量不同，则集群信息一定发生了变化；
        // 如果集群数量相同，则比较是否存在差异；如果有差异，则集群涉及节点更改
        // 并且所有接收者都需要收到节点更改事件的通知
        boolean hasChange = members.size() != serverList.size();
        // 用于存储所有新节点
        ConcurrentSkipListMap<String, Member> tmpMap = new ConcurrentSkipListMap<>();
        // 用于存储所有 UP 节点的信息
        Set<String> tmpAddressInfo = new ConcurrentHashSet<>();
        for (Member member : members) {
            final String address = member.getAddress();

            // 除了自身节点，其余节点状态在这时候全部置为 DOWN
            if (!serverList.containsKey(address)) {
                hasChange = true;
                // 如果cluster.conf或address-server中的集群信息发生了变化，
                // 在对应的tpm-server还没有启动的情况下，成员的状态应该设置为DOWN。
                // 如果对应的tpm-server已经启动，几秒后检测到成员状态会设置为UP。
                member.setState(NodeState.DOWN);
            } else {
                member.setState(serverList.get(address).getState());
            }

            // 确保节点只创建一次
            tmpMap.put(address, member);
            // 筛选出节点状态为 UP 的节点
            if (NodeState.UP.equals(member.getState())) {
                tmpAddressInfo.add(address);
            }
        }
        // 替换原有的集合
        serverList = tmpMap;
        memberAddressInfos = tmpAddressInfo;

        // 获取包括自身节点在内的所有集群节点
        Collection<Member> finalMembers = allMembers();

        log.warn("[serverlist] updated to : {}", finalMembers);

        // 将当前集群节点信息持久化到cluster.conf
        // <important> 需要把事件发布放到一个同步块中，保证事件发布是顺序的
        if (hasChange) {
            MemberUtil.syncToFile(finalMembers);
            // 发布 MembersChangeEvent 事件
            Event event = MembersChangeEvent.builder().members(finalMembers).build();
            NotifyCenter.publishEvent(event);
        }
        return hasChange;
    }

    /**
     * 成员加入这个集群。
     *
     * @param members {@link Collection} new members
     * @return is success
     */
    public synchronized boolean memberJoin(Collection<Member> members) {
        Set<Member> set = new HashSet<>(members);
        // 将新成员和老成员的信息进行汇总
        set.addAll(allMembers());
        // 判断集群节点是否有变更，
        // 如果有变更更新内存中节点信息，同时持久化新节点信息持久化到cluster.conf，发布 MembersChangeEvent 事件
        return memberChange(set);
    }

    /**
     * 成员离开这个集群。
     *
     * @param members 等待离开成员
     * @return 成员是否成功脱离集群
     */
    public synchronized boolean memberLeave(Collection<Member> members) {
        Set<Member> set = new HashSet<>(allMembers());
        set.removeAll(members);
        return memberChange(set);
    }

    /**
     * 集群成员状态是否健康
     *
     * @param address ip:port
     * @return is health
     */
    public boolean isUnHealth(String address) {
        Member member = serverList.get(address);
        if (member == null) {
            return false;
        }
        return !NodeState.UP.equals(member.getState());
    }

    public boolean isFirstIp() {
        return Objects.equals(serverList.firstKey(), this.localAddress);
    }

    @Override
    public void onApplicationEvent(WebServerInitializedEvent event) {
        // 设置自身节点状态为健康
        getSelf().setState(NodeState.UP);
        // 如果不是单机模式启动，会执行 infoReportTask，向游标指向的节点上报自己的服务
        if (!EnvUtil.getStandaloneMode()) {
            GlobalExecutor.scheduleByCommon(this.infoReportTask, 5_000L);
        }
        EnvUtil.setPort(event.getWebServer().getPort());
        EnvUtil.setLocalAddress(this.localAddress);
        log.info("此节点已准备好对外提供服务！");
    }

    /**
     * ServerMemberManager shutdown.
     *
     * @throws ServerProcessException serverProcessException
     */
    @PreDestroy
    public void shutdown() throws ServerProcessException {
        serverList.clear();
        memberAddressInfos.clear();
        infoReportTask.shutdown();
        LookupFactory.destroy();
    }

    public Set<String> getMemberAddressInfos() {
        return memberAddressInfos;
    }

    public Map<String, Member> getServerList() {
        return Collections.unmodifiableMap(serverList);
    }

    public boolean isInIpList() {
        return isInIpList;
    }

    @Override
    public String toString() {
        return "ServerMemberManager{" +
                "asyncRestTemplate=" + asyncRestTemplate +
                ", serverList=" + serverList +
                ", isInIpList=" + isInIpList +
                ", port=" + port +
                ", localAddress='" + localAddress + '\'' +
                ", lookup=" + lookup +
                ", self=" + self +
                ", memberAddressInfos=" + memberAddressInfos +
                ", infoReportTask=" + infoReportTask +
                '}';
    }

    /**
     * 同步节点元数据信息，还附上了目标节点的健康检查
     */
    class MemberInfoReportTask extends Task {

        private final GenericType<RestResult<String>> reference = new GenericType<RestResult<String>>() {
        };

        /**
         * 游标：会轮询的指向集群中除了自身以外的所有其他节点
         */
        private int cursor = 0;

        @Override
        protected void executeBody() {

            // 返回除自己以外所有集群成员。
            List<Member> members = ServerMemberManager.this.allMembersWithoutSelf();

            if (members.isEmpty()) {
                return;
            }

            // 向集群中除了自己以外的所有节点轮询广播自己的节点信息
            this.cursor = (this.cursor + 1) % members.size();
            Member target = members.get(cursor);

            log.debug("report the metadata to the node : {}", target.getAddress());

            // /tpm/v1/core/cluster/report
            final String url = HttpUtils
                    .buildUrl(false, target.getAddress(), EnvUtil.getContextPath(), Commons.TPM_CORE_CONTEXT,
                            "/cluster/report");

            try {
                // 构建请求头
                Header header = Header.newInstance().addParam(Constants.TPM_SERVER_HEADER, VersionUtils.version);
                // 将身份信息添加到 Http 头。
                AuthHeaderUtil.addIdentityToHeader(header);
                // 向目标节点发送异步请求，带上自己的 Member 节点信息
                asyncRestTemplate
                        .post(url, header,
                                Query.EMPTY, getSelf(), reference.getType(), new Callback<String>() {
                                    @Override
                                    public void onReceive(RestResult<String> result) {
                                        if (result.getCode() == HttpStatus.NOT_IMPLEMENTED.value()
                                                || result.getCode() == HttpStatus.NOT_FOUND.value()) {
                                            log.warn(">>> 😒 {} version is too low, it is recommended to upgrade the version : {}", target, VersionUtils.version);
                                            return;
                                        }
                                        if (result.ok()) {
                                            MemberUtil.onSuccess(ServerMemberManager.this, target);
                                        } else {
                                            log.warn(">>> 😒 failed to report new info to target node : {}, result : {}", target.getAddress(), result);
                                            MemberUtil.onFail(ServerMemberManager.this, target);
                                        }
                                    }

                                    @Override
                                    public void onError(Throwable throwable) {
                                        log.error(">>> 😒 failed to report new info to target node : {}, error : {}", target.getAddress(),
                                                ExceptionUtil.getAllExceptionMsg(throwable));
                                        MemberUtil.onFail(ServerMemberManager.this, target, throwable);
                                    }

                                    @Override
                                    public void onCancel() {

                                    }
                                });
            } catch (Throwable e) {
                e.printStackTrace();
                log.error("failed to report new info to target node : {}, error : {}", target.getAddress(), ExceptionUtil.getAllExceptionMsg(e));
            }
        }

        /**
         * 在 executeBody() 执行完毕后，每延时2s会循环执行当前任务
         */
        @Override
        protected void after() {
            GlobalExecutor.scheduleByCommon(this, 2_000L);
        }
    }
}
